package com.smile.service.config.mqtt;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.smile.kafka.MessageConsumer;
import com.smile.service.config.mqtt.components.MqttTopicComponent;
import com.smile.service.config.mqtt.constant.JSONConstant;
import com.smile.service.config.mqtt.constant.MethodConstant;
import com.smile.service.config.mqtt.constant.TopicConstant;
import com.smile.service.config.mqtt.constant.TopicPlaceHolderConstant;
import com.smile.service.config.mqtt.log.drc.DrcUpLogComponent;
import com.smile.service.config.mqtt.log.event.EventLogComponent;
import com.smile.service.config.mqtt.log.osd.OsdLogComponent;
import com.smile.service.config.mqtt.log.requests.RequestLogComponent;
import com.smile.service.config.mqtt.log.services.ServicesReplyLogComponent;
import com.smile.service.config.mqtt.manager.BusinessManager;
import com.smile.service.config.mqtt.utils.TopicUtils;
import com.smile.service.config.mqtt.handler.*;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;

import javax.annotation.Resource;
import java.security.NoSuchAlgorithmException;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;

@Configuration
@Slf4j
@RefreshScope
public class MqttConfig {

    @Value("${spring.mqtt.server-url}")
    private String brokerUrl;

    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    //    @Value("${spring.mqtt.client-id}")
    private String clientId = "test-nhb-123456";

    @Value("${spring.mqtt.topic}")
    private String topics;

    @Autowired
    private MqttTopicComponent mqttTopicComponent;

//    @Autowired
//    private BusinessManager businessManager;

    @Autowired
    private AbstractAirportBindStatusHandler abstractAirportBindStatusHandler;

    @Autowired
    private AbstractFlighttaskResourceGet abstractFlighttaskResourceGet;

    @Autowired
    private AbstractAirportOrganizationGetHandler abstractAirportOrganizationGetHandler;


    @Autowired
    private AbstractConfigHandler abstractConfigHandler;

    @Autowired
    private AbstractAirportOrganizationBindHandler abstractAirportOrganizationBindHandler;

    @Autowired
    private AbstractUpdateTopoHandler abstractUpdateTopoHandler;

    @Autowired
    private AbstractFlyToPointProgressHandler abstractFlyToPointProgressHandler;

    @Autowired
    private AbstractStorageConfigGetHandler abstractStorageConfigGetHandler;

    @Autowired
    private AbstractFlightAreasGetHandler abstractFlightAreasGetHandler;


//    /**
//     * kafka模板 String消息
//     */
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    MessageConsumer messageConsumer;
    @Value("${spring.mqtt.enabled}")
    private Boolean enables;

//    @Bean
//    public MqttClient mqttClient() throws MqttException {
//        MqttClient mqttClient = new MqttClient(brokerUrl, "feisuan-mqtt-producer-client", new MemoryPersistence());
//        MqttConnectOptions connOpts = new MqttConnectOptions();
//        connOpts.setUserName(username);
//        connOpts.setPassword(password.toCharArray());
//        mqttClient.connect(connOpts);
//        return mqttClient;
//    }


    // 创建MQTT客户端工厂
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{brokerUrl});
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setCleanSession(false);
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(30);
        factory.setConnectionOptions(options);
        return factory;
    }

//    // 创建出站通道（发送消息）
//    @Bean
//    public MessageChannel mqttOutboundChannel() {
//        return new DirectChannel();
//    }
//
//    // 出站消息处理器（消息生产者）
//    @Bean
//    @ServiceActivator(inputChannel = "mqttOutboundChannel")
//    public MessageHandler mqttOutbound() {
//        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId + "-sender-" + UUID.randomUUID().toString(), mqttClientFactory());
//        messageHandler.setAsync(true);
//        messageHandler.setDefaultTopic("springboot/test");
//        return messageHandler;
//    }

    // 创建入站通道（接收消息）
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    // 入站通道适配器（消息消费者）
    @Bean
    public MessageProducer inbound() {
        TopicConstant.INIT_TOPIC_LIST();


        List<String> deviceSnTopicList = getDeviceSnTopicList(mqttTopicComponent.getDeviceSnList());
        List<String> gatewaySnTopicList = getGatewaySnTopicList(mqttTopicComponent.getGatewaySnList());

        List<String> list = new ArrayList<>();
        list.addAll(deviceSnTopicList);
        list.addAll(gatewaySnTopicList);


        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter("pro-" + System.currentTimeMillis() + "-receiver", mqttClientFactory(), list.toArray(new String[0]));
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    private List<String> getGatewaySnTopicList(List<String> gatewaySnList) {

        List<String> gatewayTopicList = TopicConstant.GATEWAY_TOPIC_LIST;
        List<String> gatewaySnTopicList = new ArrayList<>();
        gatewayTopicList.forEach(i -> {
            List<String> collect = gatewaySnList.stream().map(i1 -> {
                return TopicUtils.replacePlaceholder(i, TopicPlaceHolderConstant.GATEWAY_SN, i1);
            }).collect(Collectors.toList());
            gatewaySnTopicList.addAll(collect);
        });
        return gatewaySnTopicList;
    }

    private List<String> getDeviceSnTopicList(List<String> deviceSnList) {
        List<String> deviceTopicList = TopicConstant.DEVICE_TOPIC_LIST;
        List<String> deviceSnTopicList = new ArrayList<>();
        deviceTopicList.forEach(i -> {
            List<String> collect = deviceSnList.stream().map(i1 -> {
                return TopicUtils.replacePlaceholder(i, TopicPlaceHolderConstant.DEVICE_SN, i1);
            }).collect(Collectors.toList());
            deviceSnTopicList.addAll(collect);
        });
        return deviceSnTopicList;

    }

    public String getBusinessName(String topic) {
        String[] split = topic.split("/");
        return split[split.length - 1];
    }

    /**
     * kafka模板 Message消息
     */
    @Resource
    private KafkaTemplate<String, Message<String>> kafkaTemplateMessage;


    @Resource
    private EventLogComponent eventLogComponent;

    @Resource
    private RequestLogComponent requestLogComponent;

    @Resource
    private DrcUpLogComponent drcUpComponent;

    @Resource
    private OsdLogComponent osdLogComponent;

    @Resource
    private ServicesReplyLogComponent servicesReplyLogComponent;


    @Autowired
    private AbstractEventsReplyHandler abstractEventsReplyHandler;

    // 消息处理服务（处理接收到的MQTT消息）
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) {
//                if (enables) {

                log.info("设备发来的消息为:" + message.getPayload() + " 当前时间为:" + LocalDateTime.now());

                MessageHeaders headers = message.getHeaders();

//                headers.keySet().forEach(i -> {
//                    System.out.print("key = " + i.toString());
//                    System.out.println(", value = " + headers.get(i));
//                });


//                System.out.println("headers" + headers);
                Object o = headers.get("mqtt_receivedTopic");
                String[] split = o.toString().split("/");
                String s = split[split.length - 1];

                String gateway = split[2];

                if ("osd".equals(s)) {
                    Map<String, Object> map = JSON.parseObject(message.getPayload().toString(), new TypeReference<Map<String, Object>>() {
                    });


                    map.put("topic_sn", gateway);
                    ObjectMapper objectMapper = new ObjectMapper();
                    try {
                        String jsonString = objectMapper.writeValueAsString(map);
                        messageConsumer.osd(jsonString);
//                        kafkaTemplate.send("osd", "osd",jsonString);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                    }
                    osdLogComponent.log(message.getPayload().toString());
                }

                if ("up".equals(s)) {
                    String s2 = split[split.length - 3];
                    Map<String, Object> map = JSON.parseObject(message.getPayload().toString(), new TypeReference<Map<String, Object>>() {
                    });


                    map.put("gateway", s2);
                    ObjectMapper objectMapper = new ObjectMapper();
                    try {
                        String jsonString = objectMapper.writeValueAsString(map);
                        messageConsumer.drcUp(jsonString);
//                        kafkaTemplate.send("drc_up", "drc_up_group",jsonString);
//                        kafkaTemplate.send("drc_up", "drc_up_group",jsonString);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                    }
                    drcUpComponent.log(message.getPayload().toString());
                }
//                    if ("up".equals(s)) {
//                        System.out.println("drc up的消息为:" + message.getPayload());
//                        drcUpComponent.log(message.getPayload().toString());
//                    }

                if ("connected".equals(s)) {
                    System.out.println("设备上线:" + message.getPayload());
                }

                if ("disconnected".equals(s)) {
                    System.out.println("设备下线:" + message.getPayload());
                }

                if ("events".equals(s)) {
                    try {
                        messageConsumer.event(message.getPayload().toString());
                    }catch (Exception e){

                    }

                    com.alibaba.fastjson.JSONObject jsonObject1 = com.alibaba.fastjson.JSONObject.parseObject(message.getPayload().toString());
                    String method = jsonObject1.getString("method");
//                    String gateway = jsonObject1.getString("gateway");
                    String result = jsonObject1.getString(JSONConstant.DATA);
                    if (jsonObject1.containsKey("need_reply")){
                        Integer needReply = jsonObject1.getInteger("need_reply");
                        if (needReply == 1){
                            try {
                                abstractEventsReplyHandler.process(result, method, jsonObject1.getString("tid"), jsonObject1.getString("bid"), gateway);
                            }catch (Exception e){
                                throw new RuntimeException(e);
                            }
                        }
                    }
                }

                if ("state".equals(s)) {
                    Map<String, Object> map = JSON.parseObject(message.getPayload().toString(), new TypeReference<Map<String, Object>>() {
                    });


                    map.put("topic_sn", gateway);
                    ObjectMapper objectMapper = new ObjectMapper();
                    try {
                        String jsonString = objectMapper.writeValueAsString(map);
                        messageConsumer.osd(jsonString);
//                        kafkaTemplate.send("osd", "osd",jsonString);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                    }
                }

                if ("services_reply".equals(s)) {
                    com.alibaba.fastjson.JSONObject jsonObject = com.alibaba.fastjson.JSONObject.parseObject(message.getPayload().toString());
                    String method = jsonObject.getString("method");
                    System.out.println(message.getPayload().toString());
                    System.out.println(method);
                    Map<String, Object> map = JSON.parseObject(message.getPayload().toString(), new TypeReference<Map<String, Object>>() {
                    });
                    if (method.equals("flighttask_prepare")){

                    }
                    messageConsumer.servicesReply(map,gateway);
                    servicesReplyLogComponent.log(message.getPayload().toString());
                }

                if ("requests".equals(s)
                        || "status".equals(s)
//                        || "osd".equals(s)
                ) {
                    requestLogComponent.log(message.getPayload().toString());
                    com.alibaba.fastjson.JSONObject jsonObject = com.alibaba.fastjson.JSONObject.parseObject(message.getPayload().toString());
                    String method = jsonObject.getString("method");
                    String result = jsonObject.getString(JSONConstant.DATA);
                    switch (method) {
                        case MethodConstant.AIRPORT_BIND_STATUS:
                            try {
                                abstractAirportBindStatusHandler.process(result, jsonObject.getString("tid"), jsonObject.getString("bid"), gateway);
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }
                            break;
                        case MethodConstant.AIRPORT_ORGANIZATION_GET:
                            try {
                                abstractAirportOrganizationGetHandler.process(result, jsonObject.getString("tid"), jsonObject.getString("bid"), gateway);
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }
                            break;
                        case MethodConstant.AIRPORT_ORGANIZATION_BIND:
                            try {
                                System.out.println("绑定设备");
                                abstractAirportOrganizationBindHandler.process(result, jsonObject.getString("tid"), jsonObject.getString("bid"), gateway);
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }
                            break;
                        case MethodConstant.CONFIG:
                            try {
                                abstractConfigHandler.process(result, jsonObject.getString("tid"), jsonObject.getString("bid"), gateway);
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }
                            break;
                        case MethodConstant.FLIGHT_AREAS_GET:

                            try {
                                abstractFlightAreasGetHandler.process(result, jsonObject.getString("tid"), jsonObject.getString("bid"), gateway);
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }
                            break;

                        case MethodConstant.UPDATE_TOPO:
                            try {
                                abstractUpdateTopoHandler.process(result, jsonObject.getString("tid"), jsonObject.getString("bid"), gateway);
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }
                            break;
                        case MethodConstant.STORAGE_CONFIG_GET:
                            try {
                                abstractStorageConfigGetHandler.process(result, jsonObject.getString("tid"), jsonObject.getString("bid"), gateway);
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            } catch (NoSuchAlgorithmException e) {
                                throw new RuntimeException(e);
                            }
                            break;

                        case MethodConstant.FLIGHTTASK_RESOURCE_GET:
                            try {
                                System.out.println("获取航线文件");
                                abstractFlighttaskResourceGet.process(result, jsonObject.getString("tid"), jsonObject.getString("bid"), gateway);
                                break;
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }

                    }
                }

            }
        };
    }
}